package com.zhao.demo.unbound.demo04_transformation.sample06_union

import com.zhao.demo.Raytek
import org.apache.flink.streaming.api.scala.{DataStream, SplitStream, StreamExecutionEnvironment}

/**
 * Description: 要求使用union算子,将上述的两种类型的DataStream合并起来,进行统一处理,并显示结果<br/>
 * Copyright (c) ，2020 ， 赵 <br/>
 * This program is protected by copyright laws. <br/>
 * Date： 2020/12/9 9:54
 *
 * @author 柒柒
 * @version : 1.0
 */

object StreamUnionDemo {
  def main(args: Array[String]): Unit = {
    //1.环境
    val env: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment

    //2.实时读取流数据，计算，并显示结果
    import org.apache.flink.api.scala._

    val splitStream: SplitStream[Raytek] = env.socketTextStream("node01", 7775)
      .map(perInfo => {
        val arr: Array[String] = perInfo.split(",")
        val id: String = arr(0).trim
        val temperature: Double = arr(1).trim.toDouble
        val name: String = arr(2).trim
        val timestamp: Long = arr(3).trim.toLong
        val location: String = arr(4).trim
        Raytek(id, temperature, name, timestamp, location)
      }).split((raytek: Raytek) => if (raytek.temperature >= 36.3 && raytek.temperature <= 37.2)
      Seq("正常") else Seq("异常"))

    //针对不同特征的旅客，进行不同的处理
    //a)从流中取出所有体温正常的旅客信息
    val normalTravellers: DataStream[Raytek] = splitStream.select("正常")

    //b)从流中取出所有体温异常的旅客信息
    val execptionTravellers: DataStream[Raytek] = splitStream.select("异常")

    //使用union将两种不同的流合并起来,处理后,显示结果
    normalTravellers.union(execptionTravellers).print("合并后->")

    //3.启动
    env.execute()
  }
}

















